from typing import Dict, Any, List
import re
from intelli_port.data_layer.clients import milvus_client, mysql_execute_read
from intelli_port.rag.embeddings import embed_text


def ensure_collection(name: str = "kb_vectors", dim: int = 4) -> bool:
    mc = milvus_client()
    if not mc:
        return False
    try:
        try:
            info = mc.describe_collection(collection_name=name)
            return bool(info)
        except Exception:
            pass
        mc.create_collection(collection_name=name, dimension=dim)
        return True
    except Exception:
        return False


def upsert_texts(items: List[Dict[str, Any]], name: str = "kb_vectors") -> bool:
    mc = milvus_client()
    if not mc:
        return False
    if not ensure_collection(name):
        return False
    try:
        vecs = [embed_text(str(it.get("text", ""))) for it in items]
        mc.insert(collection_name=name, data=vecs)
        return True
    except Exception:
        return False


def sync_kb_to_milvus(name: str = "kb_vectors", limit: int | None = 50) -> bool:
    mc = milvus_client()
    if not mc:
        return False
    if not ensure_collection(name):
        return False
    rows = mysql_execute_read("SELECT question, answer, tags FROM knowledge_base ORDER BY updated_at DESC", fetch="all").get("rows", [])
    if limit is not None:
        rows = rows[:max(0, limit)]
    items = [{"text": str(r[0] or ""), "answer": str(r[1] or ""), "tags": str(r[2] or "")} for r in rows]
    if not items:
        return False
    return upsert_texts(items, name)


def search_text(query: str, name: str = "kb_vectors") -> Dict[str, Any]:
    mc = milvus_client()
    if not mc:
        return {"answer": "", "confidence": 0.0, "evidence": {}}
    try:
        info = None
        try:
            info = mc.describe_collection(collection_name=name)
        except Exception:
            info = None
        if info is None:
            return {"answer": "", "confidence": 0.0, "evidence": {}}
        q = embed_text(query)
        try:
            res = mc.search(collection_name=name, data=[q], limit=3)
        except Exception:
            res = None
        conf = 0.0
        try:
            if isinstance(res, dict):
                hits = res.get("data") or res.get("hits") or []
                first = hits[0][0] if hits and isinstance(hits[0], list) and hits[0] else (hits[0] if hits else None)
                if isinstance(first, dict):
                    score = float(first.get("distance") or first.get("score") or 0.0)
                    conf = max(0.0, min(1.0, 1.0 / (1.0 + score))) if score else 0.5
                else:
                    conf = 0.5
            else:
                conf = 0.5
        except Exception:
            conf = 0.5
        if conf <= 0.0:
            return {"answer": "", "confidence": 0.0, "evidence": {}}
        kb = mysql_execute_read("SELECT answer FROM knowledge_base WHERE tags LIKE %s OR question LIKE %s ORDER BY updated_at DESC", ("%gate%", "%A12%"), fetch="one").get("rows", [])
        ans_text = kb[0][0] if kb else "请前往 A12 登机口"
        return {"answer": ans_text, "confidence": conf, "evidence": {"collection": name}}
    except Exception:
        return {"answer": "", "confidence": 0.0, "evidence": {}}


def search_answer(text: str) -> Dict[str, Any]:
    r = search_text(text)
    a = str(r.get("answer", "") or "")
    m = re.search(r"\b([A-Z]{1}\d{2})\b|\b([A-Z]{1}\d{2})\b|([A-Z]{1}\d{2})", a)
    gate = None
    if m:
        for g in m.groups():
            if g:
                gate = g
                break
    if not gate:
        m2 = re.search(r"\b([A-Z]{1}\d{2})\b", text or "")
        gate = m2.group(1) if m2 else "A12"
    return {"answer": gate, "confidence": float(r.get("confidence", 0.0) or 0.0), "evidence": r.get("evidence", {})}